package com.lzwk.app.dws.ocpx;

import com.lzwk.common.ocpx.OCPXConstant;

import java.util.HashMap;
import java.util.Map;

/**
 * @title: DwsOcpxWxappMpdailyReportgroupdata
 * @Author Yeaser
 * @Date: 2022/2/14 10:15
 * @Version 1.0
 * Description: TODO
 */
public class DwsLookUpJoinTables {

    public static  Map<String, String> dwsOcpxWxappMpdailyReportgroupdata(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_mp");

        String sourceTableMp = "create table mysql_wxapp_mpdailyreportgroupdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `date` TIMESTAMP(0), " +
                " `campaign_id` STRING, " +
                " `ad_group_id` STRING, " +
                " `cost` INT, " +
                " `view_count` INT, " +
                " `valid_click_count` INT, " +
                " `ctr` DOUBLE, " +
                " `conversions_count` INT, " +
                " `conversions_cost` INT, " +
                " `conversions_rate` DOUBLE, " +
                " `official_account_follow_count` INT, " +
                " `official_account_follow_cost` INT, " +
                " `official_account_follow_rate` DOUBLE, " +
                " `order_count` INT, " +
                " `order_rate` DOUBLE, " +
                " `order_roi` DOUBLE, " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'server-id'='35001-35010', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCWXAPPMPDAILYGROUPDATA +"', " +
                //"   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")";


        String sinkTableMp = "create table dws_ocpx_wxapp_mpdailyreportgroupdata(" +
                "   `dt` TIMESTAMP(0), " +
                "   `account_info_id` INT, " +
                "   `appid` STRING, " +
                "   `ad_group_id` STRING, " +
                "   `ad_group_name` STRING, " +
                "   `campaign_id` STRING, " +
                "   `campaign_name` STRING, " +
                "   `account_id` STRING, " +
                "   `account_name` STRING, " +
                "   `auth_info_id` INT, " +
                "   `org_id` INT, " +
                "   `cost` INT, " +
                "   `view_count` INT, " +
                "   `valid_click_count` INT, " +
                "   `ctr` double, " +
                "   `order_count` INT, " +
                "   `order_rate` double,       " +
                "   `order_roi` double, " +
                "   `conversions_count` INT, " +
                "   `conversions_cost` INT, " +
                "   `conversions_rate` double, " +
                "   `official_account_follow_count` INT, " +
                "   `official_account_follow_cost` INT, " +
                "   `official_account_follow_rate` double, " +
                "   `id` INT,  " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSWXAPPMPDAILYGROUPDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableMp = " insert into dws_ocpx_wxapp_mpdailyreportgroupdata " +
                " select " +
                " `t1`.`date` as dt, " +
                " `t1`.`account_info_id`, " +
                " ifnull(`t6`.`appid`,t5.`state`) as `appid`, " +
                " `t1`.`ad_group_id`, " +
                " regexp_replace(`t2`.`ad_group_name`, '\\n|\\t' , '') as ad_group_name, " +
                " `t1`.`campaign_id`, " +
                " regexp_replace(`t3`.`campaign_name`, '\\n|\\t' , '') as campaign_name, " +
                " `t4`.`account_id`," +
                " regexp_replace(`t4`.`account_name`, '\\n|\\t' , '') as account_name, " +
                " `t4`.`auth_info_id`, " +
                " ifnull(`t7`.`org_id`,`t8`.`org_id`) as `org_id`, " +
                " `t1`.`cost`, " +
                " `t1`.`view_count`, " +
                " `t1`.`valid_click_count`, " +
                " `t1`.`ctr`, " +
                " `t1`.`order_count`, " +
                " `t1`.`order_rate`, " +
                " `t1`.`order_roi`, " +
                " `t1`.`conversions_count`, " +
                " `t1`.`conversions_cost`, " +
                " `t1`.`conversions_rate`, " +
                " `t1`.`official_account_follow_count`, " +
                " `t1`.`official_account_follow_cost`, " +
                " `t1`.`official_account_follow_rate`, " +
                " `t1`.`id` " +
                " from mysql_wxapp_mpdailyreportgroupdata as t1 " +
                " left join ocpx_wxapp_mpadgroupdata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "   on t1.ad_group_id = t2.ad_group_id " +
                "   and t1.account_info_id = t2.account_info_id " +
                " left join ocpx_wxapp_mpcampaigndata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "   on t1.campaign_id = t3.campaign_id " +
                "   and t2.account_info_id = t3.account_info_id " +
                " left join ocpx_wxapp_mpaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                "   on t1.account_info_id = t4.id " +
                " left join ocpx_wxapp_qqauthorizationinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5" +
                "   on t4.auth_info_id = t5.id" +
                " left join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t6 " +
                "   on t5.state = t6.appid " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t7" +
                "   on t6.id = t7.object_id " +
                "   and t7.object_type = 'wxapp' " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                "   on t5.id = t8.object_id " +
                "   and t8.object_type='qq_oauth'" +
                "";

/*        tenv
                .toRetractStream(table, DwsWxappMpdailyReportgroupdata.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .addSink(MyKafkaUtil.getKafkaProducer(OCPXConstant.DWSOCPXWXAPPMPDAILYREPORTFROUPDATA,""));*/



        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableMp);
        res.put("sink",sinkTableMp);
        res.put("insert",insertTableMp);

        return res;
    }


    public static  Map<String, String> dwsOcpxWxappOceancreativereportdata(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_dy");
        String sourceTableDouyin = "create table mysql_wxapp_oceancreativereportdata(" +
                " `id` INT, " +
                " `creative_id` STRING, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `cost` DOUBLE, " +
                " `show` DOUBLE, " +
                " `avg_show_cost` DOUBLE, " +
                " `click` DOUBLE, " +
                " `avg_click_cost` DOUBLE, " +
                " `ctr` DOUBLE, " +
                " `convert` DOUBLE, " +
                " `convert_cost` DOUBLE, " +
                " `convert_rate` DOUBLE, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `creative_info_id` INT, " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'server-id'='35011-35020', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCWXAPPOCEANCREATIVEDATA +"', " +
                //"   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")";


        String sinkTableDouyin = "create table dws_ocpx_wxapp_oceancreativereportdata(" +
                "   `dt` TIMESTAMP(0), " +
                "   `creative_id` STRING, " +
                "   `org_id` INT, " +
                "   `creative_title` STRING, " +
                "   `ad_plan_id` STRING, " +
                "   `ad_plan_name` STRING, " +
                "   `campaign_id` STRING, " +
                "   `campaign_name` STRING, " +
                "   `account_id` STRING, " +
                "   `account_name` STRING, " +
                "   `auth_info_id` INT, " +
                "   `creative_info_id` INT, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `avg_show_cost` double, " +
                "   `click` double, " +
                "   `avg_click_cost` double, " +
                "   `ctr` double, " +
                "   `convert` double, " +
                "   `convert_cost` double, " +
                "   `convert_rate` double, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSWXAPPOCEANCREATIVEDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableDouyin = " insert into dws_ocpx_wxapp_oceancreativereportdata " +
                " select " +
                "   t1.`stat_datetime` AS `dt`, " +
                "   t1.`creative_id`, " +
                "   ifnull(t6.`org_id`,t8.`org_id`) as `org_id`, " +
                "   regexp_replace(t2.`title`,'\\n|\\t','') AS `creative_title`, " +
                "   t2.`ad_plan_id`, " +
                "   regexp_replace(t3.`name`,'\\n|\\t','') AS `ad_plan_name`, " +
                "   t3.`campaign_id`, " +
                "   regexp_replace(t4.`name`,'\\n|\\t','') AS `campaign_name`, " +
                "   t5.`account_id`, " +
                "   regexp_replace(t5.`account_name`,'\\n|\\t','') AS `account_name`, " +
                "   t5.`auth_info_id`, " +
                "   t1.`creative_info_id`, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`avg_show_cost`, " +
                "   t1.`click`, " +
                "   t1.`avg_click_cost`, " +
                "   t1.`ctr`, " +
                "   t1.`convert`, " +
                "   t1.`convert_cost`, " +
                "   t1.`convert_rate`, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id`  " +
                " from mysql_wxapp_oceancreativereportdata as t1 " +
                " left join ocpx_wxapp_oceancreativeinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "   on t1.creative_id = t2.creative_id " +
                " left join ocpx_wxapp_oceanadplandata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "   on t2.ad_plan_id = t3.ad_plan_id " +
                "   and t2.account_info_id = t3.account_info_id " +
                " left join ocpx_wxapp_oceancampaigndata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                "   on t3.campaign_id = t4.campaign_id " +
                "   and t3.account_info_id = t4.account_info_id " +
                " left join ocpx_wxapp_oceanaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5" +
                "   on t2.account_info_id = t5.id" +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t6" +
                "   on t5.id = t6.object_id " +
                "   and t6.object_type = 'ocean_account' " +
                " left join ocpx_wxapp_oceanauthorizationinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t7 " +
                "   on t5.auth_info_id = t7.id " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                "   on t7.id = t8.object_id " +
                "   and t8.object_type='ocean_account'" +
                "";

/*        tenv
                .toRetractStream(table, DwsWxappOceancreativereportdata.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .addSink(MyKafkaUtil.getKafkaProducer(OCPXConstant.DWSOCPXWXAPPOCEANCREATIVEREPORTDATA,""));*/

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableDouyin);
        res.put("sink",sinkTableDouyin);
        res.put("insert",insertTableDouyin);

        return res;
    }



    public static Map<String, String> dwsOcpxKsclUniytreportdata(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_ks");
        String sourceTableKs = "create table mysql_kscl_unit_report_data(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `campaign_id` STRING, " +
                " `campaign_name` STRING, " +
                " `unit_id` STRING, " +
                " `unit_name` STRING, " +
                " `cost` DOUBLE, " +
                " `show` DOUBLE, " +
                " `click` DOUBLE, " +
                " `aclick` DOUBLE, " +
                " `bclick` DOUBLE, " +
                " `photo_click_ratio` DOUBLE, " +
                " `photo_click_cost` DOUBLE, " +
                " `action_ratio` DOUBLE, " +
                " `action_cost` DOUBLE, " +
                " `avg_cost` DOUBLE, " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCKSCLUNITREPORTDATA +"', " +
                "   'server-id'='35021-35030', " +
                //"   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")";


        String sinkTableKs = "create table dws_ocpx_kscl_unit_report_data(" +
                "   `dt` TIMESTAMP(0), " +
                "   `unit_id` STRING, " +
                "   `account_info_id` INT, " +
                "   `org_id` INT,  " +
                "   `campaign_id` STRING, " +
                "   `auth_info_id` INT,  " +
                "   `account_id` STRING,  " +
                "   `account_name` STRING,  " +
                "   `campaign_name` STRING, " +
                "   `unit_name` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `aclick` double, " +
                "   `bclick` double, " +
                "   `photo_click_ratio` double, " +
                "   `photo_click_cost` double, " +
                "   `action_ratio` double, " +
                "   `action_cost` double, " +
                "   `avg_cost` double, " +
                "   `id` INT, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSKSCLUNITREPORTDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableKs = "insert into dws_ocpx_kscl_unit_report_data " +
                " select " +
                "  t1.`stat_datetime` AS `dt`, " +
                "  t1.`unit_id`, " +
                "  t1.`account_info_id`, " +
                "  t3.`org_id`,  " +
                "  t1.`campaign_id`, " +
                "  t2.`auth_info_id`,  " +
                "  t2.`account_id`,  " +
                "  regexp_replace(t2.`account_name`,'\\n|\\t',''),  " +
                "  regexp_replace(t1.`campaign_name`,'\\n|\\t',''),  " +
                "  regexp_replace(t1.`unit_name`,'\\n|\\t',''),  " +
                "  t1.`cost`, " +
                "  t1.`show`, " +
                "  t1.`click`, " +
                "  t1.`aclick`, " +
                "  t1.`bclick`, " +
                "  t1.`photo_click_ratio`, " +
                "  t1.`photo_click_cost`, " +
                "  t1.`action_ratio`, " +
                "  t1.`action_cost`, " +
                "  t1.`avg_cost`, " +
                "  t1.`id`  " +
                " from mysql_kscl_unit_report_data as t1 " +
                " left join ocpx_platform_account_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "   on t1.account_info_id = t2.id " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "   on t2.auth_info_id = t3.object_id " +
                "   and t3.object_type = 'platform_auth' " +
                "";

/*        tenv
                .toRetractStream(table, DwsKsclUnitreportdata.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .addSink(MyKafkaUtil.getKafkaProducer(OCPXConstant.DWSOCPXKSCLUNITREPORTDATA,""));*/

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableKs);
        res.put("sink",sinkTableKs);
        res.put("insert",insertTableKs);

        return res;

    }


    public static Map<String, String> dwsOcpxWxAppOceanGenderAudienceReportData(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_gender");
        String sourceTableGender = "create table mysql_wxapp_oceangenderaudiencereportdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `advertiser_id` STRING, " +
                " `gender_name` STRING, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `cost` double, " +
                " `show` double, " +
                " `click` double, " +
                " `download_finish` double, " +
                " `convert` double, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCWXAPPGENDERAUDIENCEDATA +"', " +
                "   'server-id'='35031-35040', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")";


        String sinkTableGender = "create table dws_ocpx_wxapp_oceangenderaudiencereportdata(" +
                "   `account_info_id` INT, " +
                "   `stat_datetime` TIMESTAMP(0), " +
                "   `gender_name` STRING, " +
                "   `org_id` INT, " +
                "   `advertiser_id` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `download_finish` double, " +
                "   `convert` double, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSWXAPPGENDERAUDIENCEDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableGender = "insert into dws_ocpx_wxapp_oceangenderaudiencereportdata " +
                " select " +
                "   t1.`account_info_id`, " +
                "   t1.`stat_datetime`, " +
                "   t1.`gender_name`, " +
                "   ifnull(t3.org_id,t5.org_id)  as org_id, " +
                "   t1.`advertiser_id`, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`click`, " +
                "   t1.`download_finish`, " +
                "   t1.`convert`, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id` " +
                " from mysql_wxapp_oceangenderaudiencereportdata t1 " +
                " left join ocpx_wxapp_oceanaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "   on t1.account_info_id = t2.id  " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3  " +
                "   on t2.id = t3.object_id " +
                "   and t3.object_type='ocean_account' " +
                " left join ocpx_wxapp_oceanauthorizationinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4  " +
                "   on t2.auth_info_id = t4.id  " +
                " left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5  " +
                "   on t4.id = t5.object_id " +
                "   and t5.object_type='ocean_account' " +
                "";
        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableGender);
        res.put("sink",sinkTableGender);
        res.put("insert",insertTableGender);

        return res;

    }


    public static Map<String, String> dwsOcpxWxAppOceanAgeAudienceReportData(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_age");
        String sourceTableAge = "create table mysql_wxapp_oceanageaudiencereportdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `advertiser_id` STRING, " +
                " `age_name` STRING, " +
                " `stat_datetime`  TIMESTAMP(0), " +
                " `cost` double, " +
                " `show` double, " +
                " `click` double, " +
                " `download_finish` double, " +
                " `convert` double, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCWXAPPAGEAUDIENCEDATA +"', " +
                "   'server-id'='35041-35050', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableAge = "create table dws_ocpx_wxapp_oceanageaudiencereportdata(" +
                "   `account_info_id` INT, " +
                "   `stat_datetime` TIMESTAMP(0), " +
                "   `age_name` STRING, " +
                "   `org_id` INT, " +
                "   `advertiser_id` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `download_finish` double, " +
                "   `convert` double, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSWXAPPAGEAUDIENCEDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableAge = "insert into dws_ocpx_wxapp_oceanageaudiencereportdata " +
                " select " +
                "   t1.`account_info_id`, " +
                "   t1.`stat_datetime`, " +
                "   regexp_replace(t1.`age_name`,'\\n|\\t','') as age_name , " +
                "   ifnull(t3.org_id,t5.org_id)  as org_id, " +
                "   t1.`advertiser_id`, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`click`, " +
                "   t1.`download_finish`, " +
                "   t1.`convert`, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id` " +
                "  from mysql_wxapp_oceanageaudiencereportdata t1 " +
                "  left join ocpx_wxapp_oceanaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "   on t1.account_info_id = t2.id  " +
                "  left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3  " +
                "    on t2.id = t3.object_id " +
                "    and t3.object_type='ocean_account' " +
                "  left join ocpx_wxapp_oceanauthorizationinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4  " +
                "    on t2.auth_info_id = t4.id  " +
                "  left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5  " +
                "    on t4.id = t5.object_id " +
                "    and t5.object_type='ocean_account' " +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableAge);
        res.put("sink",sinkTableAge);
        res.put("insert",insertTableAge);

        return res;
    }

    public static Map<String, String> dwsOcpxWxAppOceanCityAudienceReportData(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_city");
        String sourceTableCity = "create table mysql_wxapp_oceancityaudiencereportdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `advertiser_id` STRING, " +
                " `province_name` STRING, " +
                " `city_name` STRING, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `cost` double, " +
                " `show` double, " +
                " `click` double, " +
                " `download_finish` double, " +
                " `convert` double, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCWXAPPCITYAUDIENCEDATA +"', " +
                "   'server-id'='35051-35060', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableCity = "create table dws_ocpx_wxapp_oceancityaudiencereportdata(" +
                "   `account_info_id` INT, " +
                "   `stat_datetime` TIMESTAMP(0), " +
                "   `province_name` STRING, " +
                "   `city_name` STRING, " +
                "   `org_id` INT, " +
                "   `advertiser_id` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `download_finish` double, " +
                "   `convert` double, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSWXAPPCITYAUDIENCEDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableCity = "insert into dws_ocpx_wxapp_oceancityaudiencereportdata " +
                " select " +
                "   t1.`account_info_id`, " +
                "   t1.`stat_datetime`, " +
                "   t1.`province_name`, " +
                "   t1.`city_name`, " +
                "   ifnull(t3.org_id,t5.org_id)  as org_id, " +
                "   t1.`advertiser_id`, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`click`, " +
                "   t1.`download_finish`, " +
                "   t1.`convert`, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id` " +
                "   from mysql_wxapp_oceancityaudiencereportdata t1 " +
                "   left join ocpx_wxapp_oceanaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "       on t1.account_info_id = t2.id  " +
                "   left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3  " +
                "       on t2.id = t3.object_id " +
                "       and t3.object_type='ocean_account' " +
                "   left join ocpx_wxapp_oceanauthorizationinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4  " +
                "       on t2.auth_info_id = t4.id  " +
                "   left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5  " +
                "       on t4.id = t5.object_id " +
                "       and t5.object_type='ocean_account' " +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableCity);
        res.put("sink",sinkTableCity);
        res.put("insert",insertTableCity);

        return res;
    }

    public static Map<String, String> dwsOcpxAdServingReport(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_city");
        String sourceTableAdServing = "create table mysql_ad_serving_report(" +
                " `id` INT, " +
                " `dim_type` STRING, " +
                " `dim_id` INT, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `show_cnt` INT, " +
                " `cost` DOUBLE, " +
                " `click_cnt` INT, " +
                " `ctr` double, " +
                " `avg_click_cost` double, " +
                " `convert` double, " +
                " `convert_cost` double, " +
                " `convert_rate` double, " +
                " `pay_order_count` INT, " +
                " `view_cnt` INT, " +
                " `watch_one_minute_cnt` INT, " +
                " `follow_cnt` INT, " +
                " `comment_cnt` INT, " +
                " `share_cnt` INT, " +
                " `detail_data` STRING, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCADSERVINGREPORT +"', " +
                "   'server-id'='35061-35070', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableAdServing = "create table dws_ocpx_ad_servingreport(" +
                " `dt` TIMESTAMP(0), " +
                " `org_id` INT, " +
                " `id` INT, " +
                " `ad_plan_id` STRING, " +
                " `ad_plan_name` STRING, " +
                " `campaign_id` STRING, " +
                " `campaigm_name` STRING, " +
                " `account_id` STRING, " +
                " `account_name` STRING, " +
                " `auth_info_id` INT, " +
                " `dim_type` STRING, " +
                " `dim_id` INT, " +
                " `show_cnt` INT, " +
                " `avg_show_cost` DOUBLE, " +
                " `luban_live_slidecart_click_cnt` INT, " +
                " `luban_live_click_product_cnt` INT, " +
                " `luban_live_gift_cnt` INT, " +
                " `luban_live_gift_amount` double, " +
                " `cost` DOUBLE, " +
                " `click_cnt` INT, " +
                " `ctr` double, " +
                " `avg_click_cost` double, " +
                " `convert` double, " +
                " `convert_cost` double, " +
                " `convert_rate` double, " +
                " `pay_order_count` INT, " +
                " `view_cnt` INT, " +
                " `watch_one_minute_cnt` INT, " +
                " `follow_cnt` INT, " +
                " `comment_cnt` INT, " +
                " `share_cnt` INT, " +
                " `detail_data` STRING, " +
                " `create_time` STRING, " +
                " `update_time` STRING, " +
                "   PRIMARY KEY (`dt`, `org_id`, `id`,`ad_plan_id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSADSERVINGREPORT + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableAdServing = "insert into dws_ocpx_ad_servingreport " +
                " select " +
                "   t1.`stat_datetime` as dt, " +
                "   t5.org_id as `org_id`, " +
                "   t1.`id`, " +
                "   t2.`ad_plan_id`, " +
                "   regexp_replace(t2.`name`,'\\n|\\t','') as ad_plan_name, " +
                "   t2.`campaign_id`, " +
                "   regexp_replace(t3.`name`,'\\n|\\t','') as campaign_name, " +
                "   t4.account_id, " +
                "   regexp_replace(t4.`account_name`,'\\n|\\t','') AS account_name, " +
                "   t4.auth_info_id, " +
                "   t1.`dim_type`, " +
                "   t1.`dim_id`, " +
                "   t1.`show_cnt`, " +
                "   GetJsonDouble(t1.detail_data,'cpm_platform',0.0) as avg_show_cost, " +
                "   GetJsonInt(t1.detail_data,'luban_live_slidecart_click_cnt',0) as luban_live_slidecart_click_cnt, " +
                "   GetJsonInt(t1.detail_data,'luban_live_click_product_cnt',0) as luban_live_click_product_cnt, " +
                "   GetJsonInt(t1.detail_data,'luban_live_gift_cnt',0) as luban_live_gift_cnt, " +
                "   GetJsonDouble(t1.detail_data,'luban_live_gift_amount',0.0) as luban_live_gift_amount, " +
                "   t1.`cost`, " +
                "   t1.`click_cnt`, " +
                "   t1.`ctr`, " +
                "   t1.`avg_click_cost`, " +
                "   t1.`convert`, " +
                "   t1.`convert_cost`, " +
                "   t1.`convert_rate`, " +
                "   t1.`pay_order_count`, " +
                "   t1.`view_cnt`, " +
                "   t1.`watch_one_minute_cnt`, " +
                "   t1.`follow_cnt`, " +
                "   t1.`comment_cnt`, " +
                "   t1.`share_cnt`, " +
                "   regexp_replace(t1.`detail_data`,'\\n|\\t','') AS detail_data, " +
                "   CAST(DATE_FORMAT(t1.create_time,'yyyy-MM-dd HH:mm:ss') AS STRING) as create_time, " +
                "   CAST(DATE_FORMAT(t1.update_time,'yyyy-MM-dd HH:mm:ss') AS STRING) as update_time " +
                "   from mysql_ad_serving_report t1 " +
                "   left join ocpx_wxapp_oceanadplandata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "       on t1.dim_id = t2.id  " +
                "   left join ocpx_wxapp_oceancampaigndata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3  " +
                "      on t2.campaign_id = t3.campaign_id " +
                "       and t2.account_info_id = t3.account_info_id " +
                "   left join ocpx_wxapp_oceanaccountinfo FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4  " +
                "       on t2.account_info_id = t4.id  " +
                "   left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5  " +
                "       on t4.id = t5.object_id " +
                "       and t5.object_type='ocean_qc' " +
                "   where t1.dim_type= 'AD_PLAN'" +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableAdServing);
        res.put("sink",sinkTableAdServing);
        res.put("insert",insertTableAdServing);

        return res;
    }



    public static Map<String, String> dwsOcpxBaiduCreativeReport(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_city");
        String sourceTableBaidu = "create table mysql_baidu_creative_report(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `campaign_id` STRING , " +
                " `campaign_name` STRING, " +
                " `unit_id` STRING, " +
                " `unit_name` STRING, " +
                " `creative_id` STRING, " +
                " `creative_name` STRING, " +
                " `cost` double, " +
                " `impression` double, " +
                " `click` double, " +
                " `ctr` double, " +
                " `cpc` double, " +
                " `cpm` double, " +
                " `target_trans` double, " +
                " `target_trans_cpc` double, " +
                " `target_trans_ratio` double, " +
                " `form_submit` double, " +
                " `form_submit_cost` double, " +
                " `paid` double, " +
                " `paid_cost` double, " +
                " `detail_data` STRING, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXBAIDUCREATIVEREPORT +"', " +
                "   'server-id'='35071-35080', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableBaidu = "create table dws_baidu_creative_report(" +
                " `dt` TIMESTAMP(0), " +
                " `account_info_id` INT, " +
                " `id` INT, " +
                " `org_id` INT, " +
                " `creative_id` STRING, " +
                " `auth_info_id` INT, " +
                " `account_id` STRING, " +
                " `account_name` STRING, " +
                " `campaigm_id` STRING, " +
                " `campaigm_name` STRING, " +
                " `unit_id` STRING, " +
                " `unit_name` STRING, " +
                " `creative_name` STRING, " +
                " `cost` DOUBLE, " +
                " `impression` DOUBLE, " +
                " `click` DOUBLE, " +
                " `ctr` DOUBLE, " +
                " `cpc` DOUBLE, " +
                " `cpm` DOUBLE, " +
                " `target_trans` DOUBLE, " +
                " `target_trans_cpc` DOUBLE, " +
                " `target_trans_ratio` DOUBLE, " +
                " `form_submit` DOUBLE, " +
                " `form_submit_cost` DOUBLE, " +
                " `paid` DOUBLE, " +
                " `paid_cost` DOUBLE, " +
                " `detail_data` STRING, " +
                " `create_time` STRING, " +
                " `update_time` STRING, " +
                "   PRIMARY KEY (`dt`, `account_info_id`, `id`,`org_id`,`creative_id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSBAIDUCREATIVEREPORT + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableBaidu = "insert into dws_baidu_creative_report " +
                " select " +
                "   t1.`stat_datetime` as dt, " +
                "   t1.account_info_id, " +
                "   t1.`id`, " +
                "   t3.org_id as `org_id`, " +
                "   t1.`creative_id`, " +
                "   t2.`auth_info_id`, " +
                "   t2.account_id as account_id, " +
                "   regexp_replace(t2.`account_name`,'\\n|\\t','') as account_name, " +
                "   t1.`campaign_id`, " +
                "   regexp_replace(t1.`campaign_name`,'\\n|\\t','') as campaign_name, " +
                "   t1.`unit_id`, " +
                "   regexp_replace(t1.`unit_name`,'\\n|\\t','') as unit_name, " +
                "   regexp_replace(t1.`creative_name`,'\\n|\\t','') AS creative_name, " +
                "   t1.`cost`, " +
                "   t1.`impression`, " +
                "   t1.`click`, " +
                "   t1.`ctr`, " +
                "   t1.`cpc`, " +
                "   t1.`cpm`, " +
                "   t1.`target_trans`, " +
                "   t1.`target_trans_cpc`, " +
                "   t1.`target_trans_ratio`, " +
                "   t1.`form_submit`, " +
                "   t1.`form_submit_cost`, " +
                "   t1.`paid`, " +
                "   t1.`paid_cost`, " +
                "   regexp_replace(t1.`detail_data`,'\\n|\\t','') AS detail_data, " +
                "   CAST(DATE_FORMAT(t1.create_time,'yyyy-MM-dd HH:mm:ss') AS STRING) as create_time, " +
                "   CAST(DATE_FORMAT(t1.update_time,'yyyy-MM-dd HH:mm:ss') AS STRING) as update_time " +
                "   from mysql_baidu_creative_report t1 " +
                "   left join ocpx_platform_account_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "       on t1.account_info_id = t2.id  " +
                "   left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3  " +
                "      on t1.account_info_id = t3.object_id " +
                "      and t3.object_type='platform_account' " +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableBaidu);
        res.put("sink",sinkTableBaidu);
        res.put("insert",insertTableBaidu);

        return res;
    }


    public static Map<String, String> dwsOcpxPlatformAgeReportData(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_city");
        String sourceTableAge = "create table mysql_platform_agereportdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `platform_type` STRING, " +
                " `advertiser_id` STRING, " +
                " `age_name` STRING, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `cost` double, " +
                " `show` double, " +
                " `click` double, " +
                " `convert` double, " +
                " `ctr` double, " +
                " `cpc` double, " +
                " `detail_data` STRING, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCPLATFORMAGEDATA +"', " +
                "   'server-id'='35091-35100', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableAge = "create table dws_ocpx_platform_age_data(" +
                "   `account_info_id` INT, " +
                "   `stat_datetime` TIMESTAMP(0), " +
                "   `age_name` STRING, " +
                "   `platform_type` STRING, " +
                "   `org_id` INT, " +
                "   `advertiser_id` STRING, " +
                "   `account_name` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `convert` double, " +
                "   `ctr` double, " +
                "   `cpc` double, " +
                "   `detail_data` STRING, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`account_info_id`,`stat_datetime`,`age_name`,`platform_type`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSPLATFORMAGEDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableAge = "insert into dws_ocpx_platform_age_data " +
                " select " +
                "   t1.`account_info_id`, " +
                "   t1.`stat_datetime`, " +
                "   regexp_replace(`t1`.`age_name`, '\\n|\\t' , '') as age_name, " +
                "   t1.`platform_type`, " +
                "   t3.org_id, " +
                "   t1.`advertiser_id`, " +
                "   regexp_replace(t2.`account_name`,'\\n|\\t','') as account_name, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`click`, " +
                "   t1.`convert`, " +
                "   t1.`ctr`, " +
                "   t1.`cpc`, " +
                "   regexp_replace(`t1`.`detail_data`, '\\n|\\t' , '') as detail_data, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id` " +
                "   from mysql_platform_agereportdata t1 " +
                "   LEFT JOIN ocpx_platform_account_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "       ON t1.account_info_id = t2.id  " +
                "   LEFT JOIN ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "       ON t2.auth_info_id = t3.object_id " +
                "       AND t3.object_type='platform_account'  " +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableAge);
        res.put("sink",sinkTableAge);
        res.put("insert",insertTableAge);

        return res;
    }


    public static Map<String, String> dwsOcpxPlatformCityReportData(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_city");
        String sourceTableCity = "create table mysql_platform_cityreportdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `platform_type` STRING, " +
                " `advertiser_id` STRING, " +
                " `province_name` STRING, " +
                " `city_name` STRING, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `cost` double, " +
                " `show` double, " +
                " `click` double, " +
                " `convert` double, " +
                " `ctr` double, " +
                " `cpc` double, " +
                " `detail_data` STRING, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCPLATFORMCITYDATA +"', " +
                "   'server-id'='35101-35110', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableCity = "create table dws_ocpx_platform_city_data(" +
                "   `account_info_id` INT, " +
                "   `stat_datetime` TIMESTAMP(0), " +
                "   `province_name` STRING, " +
                "   `city_name` STRING, " +
                "   `platform_type` STRING, " +
                "   `org_id` INT, " +
                "   `advertiser_id` STRING, " +
                "   `account_name` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `convert` double, " +
                "   `ctr` double, " +
                "   `cpc` double, " +
                "   `detail_data` STRING, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`account_info_id`,`stat_datetime`,`province_name`,`city_name`,`platform_type`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSPLATFORMCITYDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableCity = "insert into dws_ocpx_platform_city_data " +
                " select " +
                "   t1.`account_info_id`, " +
                "   t1.`stat_datetime`, " +
                "   regexp_replace(`t1`.`province_name`, '\\n|\\t' , '') as province_name, " +
                "   regexp_replace(`t1`.`city_name`, '\\n|\\t' , '') as city_name, " +
                "   t1.`platform_type`, " +
                "   t3.org_id, " +
                "   t1.`advertiser_id`, " +
                "   regexp_replace(t2.`account_name`,'\\n|\\t','') as account_name, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`click`, " +
                "   t1.`convert`, " +
                "   t1.`ctr`, " +
                "   t1.`cpc`, " +
                "   regexp_replace(`t1`.`detail_data`, '\\n|\\t' , '') as detail_data, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id` " +
                "   from mysql_platform_cityreportdata t1 " +
                "   LEFT JOIN ocpx_platform_account_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "       ON t1.account_info_id = t2.id  " +
                "   LEFT JOIN ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "       ON t1.account_info_id = t3.object_id " +
                "       AND t3.object_type='platform_account'  " +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableCity);
        res.put("sink",sinkTableCity);
        res.put("insert",insertTableCity);

        return res;
    }



    public static Map<String, String> dwsOcpxPlatformGenderReportData(){
        //tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_city");
        String sourceTableGender = "create table mysql_platform_genderreportdata(" +
                " `id` INT, " +
                " `account_info_id` INT, " +
                " `platform_type` STRING, " +
                " `advertiser_id` STRING, " +
                " `gender_name` STRING, " +
                " `stat_datetime` TIMESTAMP(0), " +
                " `cost` double, " +
                " `show` double, " +
                " `click` double, " +
                " `convert` double, " +
                " `ctr` double, " +
                " `cpc` double, " +
                " `detail_data` STRING, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` as PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='"+ OCPXConstant.OCPXSLAVEHOST +"', " +
                "   'port'='"+ OCPXConstant.OCPXSLAVEPORT +"', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME +"', " +
                "   'password'='"+ OCPXConstant.OCPXSLAVEPASSWORD +"', " +
                "   'database-name'='"+ OCPXConstant.OCPXSLAVEDBNAME +"', " +
                "   'table-name'='"+ OCPXConstant.OCPXCDCPLATFORMGENDERDATA +"', " +
                "   'server-id'='35111-35120', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '"+ OCPXConstant.SCANINCREMENTALSNAPSHOTCHUNKSIZE +"' " +
                ")";


        String sinkTableGender = "create table dws_ocpx_platform_gender_data(" +
                "   `account_info_id` INT, " +
                "   `stat_datetime` TIMESTAMP(0), " +
                "   `gender_name` STRING, " +
                "   `platform_type` STRING, " +
                "   `org_id` INT, " +
                "   `advertiser_id` STRING, " +
                "   `account_name` STRING, " +
                "   `cost` double, " +
                "   `show` double, " +
                "   `click` double, " +
                "   `convert` double, " +
                "   `ctr` double, " +
                "   `cpc` double, " +
                "   `detail_data` STRING, " +
                "   `create_time` TIMESTAMP(0), " +
                "   `update_time` TIMESTAMP(0), " +
                "   `id` INT, " +
                "   PRIMARY KEY (`account_info_id`,`stat_datetime`,`gender_name`,`platform_type`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='"+ OCPXConstant.OCPXDORISIP +"', " +
                "   'table.identifier'='"+ OCPXConstant.OCPXDORISDATABASE + "."  + OCPXConstant.OCPXDWSPLATFORMGENDERDATA + "', " +
                "   'username'='"+ OCPXConstant.OCPXDORISUSERNAME +"'," +
                "   'password'='"+ OCPXConstant.OCPXDORISPASSWORD +"'," +
                "   'sink.batch.size'='"+ OCPXConstant.OCPXDORISSINKBATCHSIZE +"', " +
                "   'sink.batch.interval'='"+ OCPXConstant.OCPXDORISSINKBATCHINTERVAL +"' " +
                ")";


        String insertTableGender = "insert into dws_ocpx_platform_gender_data " +
                " select " +
                "   t1.`account_info_id`, " +
                "   t1.`stat_datetime`, " +
                "   regexp_replace(`t1`.`gender_name`, '\\n|\\t' , '') as gender_name, " +
                "   t1.`platform_type`, " +
                "   t3.org_id, " +
                "   t1.`advertiser_id`, " +
                "   regexp_replace(t2.`account_name`,'\\n|\\t','') as account_name, " +
                "   t1.`cost`, " +
                "   t1.`show`, " +
                "   t1.`click`, " +
                "   t1.`convert`, " +
                "   t1.`ctr`, " +
                "   t1.`cpc`, " +
                "   regexp_replace(`t1`.`detail_data`, '\\n|\\t' , '') as detail_data, " +
                "   t1.`create_time`, " +
                "   t1.`update_time`, " +
                "   t1.`id` " +
                "   from mysql_platform_genderreportdata t1 " +
                "   LEFT JOIN ocpx_platform_account_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2  " +
                "       ON t1.account_info_id = t2.id  " +
                "   LEFT JOIN ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                "       ON t1.account_info_id = t3.object_id " +
                "       AND t3.object_type='platform_account'  " +
                "";

        Map<String, String> res = new HashMap<>();
        res.put("source",sourceTableGender);
        res.put("sink",sinkTableGender);
        res.put("insert",insertTableGender);

        return res;
    }

}
